package cn.itcast.task;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.IndexBean;
import cn.itcast.config.QuotConfig;
import cn.itcast.function.KeyFunction;
import cn.itcast.function.MinIndexWindowFunction;
import cn.itcast.inter.ProcessDataInterface;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Properties;

public class IndexMinTask implements ProcessDataInterface {
    /**
     * 1.定义侧边流
     * 2.数据分组
     * 3.划分时间窗口
     * 4.分时数据处理（新建分时窗口函数）
     * 5.数据分流
     * 6.数据分流转换
     * 7.分表存储(写入kafka)
     */

    @Override
    public void process(DataStream<CleanBean> waterData) {
        OutputTag<IndexBean> szseOpt = new OutputTag<>("szseOpt", TypeInformation.of(IndexBean.class));

        SingleOutputStreamOperator<IndexBean> processData = waterData
                .keyBy(new KeyFunction())
                .timeWindow(Time.seconds(60))
                .apply(new MinIndexWindowFunction())
                //分流
                .process(new ProcessFunction<IndexBean, IndexBean>() {
                    @Override
                    public void processElement(IndexBean indexBean, ProcessFunction<IndexBean, IndexBean>.Context context, Collector<IndexBean> collector) throws Exception {
                        if (indexBean.getSource().equals(QuotConfig.config.getProperty("sse.topic"))) {
                            collector.collect(indexBean);
                        } else {
                            context.output(szseOpt, indexBean);
                        }
                    }
                });

        //todo 6.数据分流转换
        //插入druid数据源时json格式，需要把最终的数据转换成json字符串
        //沪市分时行情
        SingleOutputStreamOperator<String> sseStr = processData.map(new MapFunction<IndexBean, String>() {
            @Override
            public String map(IndexBean indexBean) throws Exception {
                return JSON.toJSONString(indexBean);
            }
        });
        //深市
        SingleOutputStreamOperator<String> szseStr = processData.getSideOutput(szseOpt).map(new MapFunction<IndexBean, String>() {
            @Override
            public String map(IndexBean indexBean) throws Exception {
                return JSON.toJSONString(indexBean);
            }
        });

        Properties prop = new Properties();
        prop.setProperty("bootstrap.servers",QuotConfig.config.getProperty("bootstrap.servers"));
        FlinkKafkaProducer011<String> sseKafkaPro = new FlinkKafkaProducer011<String>(
                QuotConfig.config.getProperty("sse.index.topic"),
                new SimpleStringSchema(),
                prop
        );
        FlinkKafkaProducer011<String> szseKafkaPro = new FlinkKafkaProducer011<String>(
                QuotConfig.config.getProperty("szse.index.topic"),
                new SimpleStringSchema(),
                prop
        );

        sseStr.addSink(sseKafkaPro);
        szseStr.addSink(szseKafkaPro);
    }
}
